use std::sync::Arc;

use crate::config::Config;
use crate::consumer::{ConsumeLogs, ConsumeMetrics, ConsumeTraces};
use crate::exporter::debug::DebugExporter;
use crate::exporter::otlp_grpc_exporter::OtlpGrpcExporter;
use crate::exporter::otlp_http_exporter::OtlpHttpExporter;
use crate::command::{CommandBroadcaster};
use crate::command::service::CommandService;
use crate::pipeline::{
    LogsPipelineBuilder, MetricsPipelineBuilder, TracesPipelineBuilder,
    fanout::{FanoutMetrics, FanoutTraces},
};
use crate::processor::memory_limiter::{MemoryLimiter};
use crate::processor::batch_processor::{LogsBatchProcessor, TracesBatchProcessor, MetricsBatchProcessor};
use crate::receiver::otlp_grpc_receiver::OtlpGrpcReceiver;
use crate::receiver::otlp_http_receiver::OtlpHttpReceiver;

/// Build consumer pipelines based on configuration.
async fn build_consumers(
    cfg: &Config,
    broadcaster: Arc<CommandBroadcaster>,
) -> Result<
    (
        Arc<dyn ConsumeTraces + Send + Sync>,
        Arc<dyn ConsumeMetrics + Send + Sync>,
        Arc<dyn ConsumeLogs + Send + Sync>,
    ),
    Box<dyn std::error::Error + Send + Sync>,
> {
    let mut traces_exporters: Vec<Arc<dyn ConsumeTraces + Send + Sync>> = Vec::new();
    let mut metrics_exporters: Vec<Arc<dyn ConsumeMetrics + Send + Sync>> = Vec::new();
    let mut logs_exporters: Vec<Arc<dyn ConsumeLogs + Send + Sync>> = Vec::new();

    if cfg.exporters.debug {
        let debug_exp = DebugExporter::new(true);
        traces_exporters.push(debug_exp.clone());
        metrics_exporters.push(debug_exp.clone());
        logs_exporters.push(debug_exp.clone());
    }

    if !cfg.exporters.grpc.endpoint.is_empty() {
        let grpc_exp = OtlpGrpcExporter::connect(cfg.exporters.grpc.endpoint.clone()).await?;
        traces_exporters.push(grpc_exp.clone());
        metrics_exporters.push(grpc_exp.clone());
        logs_exporters.push(grpc_exp.clone());
    }

    if !cfg.exporters.http.endpoint.is_empty() {
        let http_exp = OtlpHttpExporter::new(cfg).await?;
        http_exp.subscribe_to_commands(broadcaster.clone());
        traces_exporters.push(http_exp.clone());
        metrics_exporters.push(http_exp.clone());
        logs_exporters.push(http_exp.clone());
    }

    let traces_exporter: Arc<dyn ConsumeTraces + Send + Sync> =
        Arc::new(FanoutTraces::new(traces_exporters));
    let metrics_exporter: Arc<dyn ConsumeMetrics + Send + Sync> =
        Arc::new(FanoutMetrics::new(metrics_exporters));
    let logs_exporter: Arc<dyn ConsumeLogs + Send + Sync> =
        Arc::new(crate::pipeline::fanout::FanoutLogs::new(logs_exporters));
    
    let memory_limiter = MemoryLimiter::new(cfg.processors.memory_limiter.clone());

    let traces_batch_processor = TracesBatchProcessor::new(cfg.processors.batch.clone());
    let metrics_batch_processor = MetricsBatchProcessor::new(cfg.processors.batch.clone());
    let logs_batch_processor = LogsBatchProcessor::new(cfg.processors.batch.clone());

    let traces_consumer = TracesPipelineBuilder::new(traces_exporter)
        .append_processor(memory_limiter.clone())
        .append_processor(traces_batch_processor.clone())
        .build();

    let metrics_consumer = MetricsPipelineBuilder::new(metrics_exporter)
        .append_processor(memory_limiter.clone())
        .append_processor(metrics_batch_processor.clone())
        .build();

    let logs_consumer = LogsPipelineBuilder::new(logs_exporter)
        .append_processor(memory_limiter.clone())
        .append_processor(logs_batch_processor.clone())
        .build();

    Ok((traces_consumer, metrics_consumer, logs_consumer))
}

/// Run the OpenTelemetry collector with the given configuration.
///
/// This function orchestrates the startup of all collector components:
/// - Creates command broadcaster for runtime control
/// - Builds consumer pipelines based on configuration
/// - Starts receivers (gRPC, HTTP, command service)
/// - Runs until interrupted
pub async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let broadcaster = Arc::new(CommandBroadcaster::new());

    let (traces_consumer, metrics_consumer, logs_consumer) = build_consumers(cfg, broadcaster.clone()).await?;

    let mut tasks = Vec::new();

    if !cfg.receivers.grpc.is_empty() {
        let grpc_receiver = OtlpGrpcReceiver::new(
            Some(traces_consumer.clone()),
            Some(metrics_consumer.clone()),
            Some(logs_consumer.clone()),
        );
        let socket_addr = cfg.receivers.grpc.parse().unwrap();
        tasks.push(tokio::spawn(
            async move { grpc_receiver.run(socket_addr).await },
        ));
    }

    // Start HTTP receiver if configured
    if !cfg.receivers.http.is_empty() {
        let http_receiver = OtlpHttpReceiver::new(
            Some(traces_consumer.clone()),
            Some(metrics_consumer.clone()),
            Some(logs_consumer.clone()),
        );
        let socket_addr = cfg.receivers.http.parse().unwrap();
        tasks.push(tokio::spawn(
            async move { http_receiver.run(socket_addr).await },
        ));
    }

    // Start Command service if configured
    if !cfg.receivers.command.is_empty() {
        let cmd_addr = cfg.receivers.command.parse().unwrap();
        let cmd_service = CommandService::new(broadcaster.clone());
        tasks.push(tokio::spawn(async move { cmd_service.run(cmd_addr).await }));
    }
    if tasks.is_empty() {
        eprintln!("No receivers configured (grpc/http). Exiting.");
        return Ok(());
    }

    tokio::select! {
        biased;
        _ = tokio::signal::ctrl_c() => {
            println!("\n🔄 Received shutdown signal...");
        }
        _ = async {
            for t in tasks { let _ = t.await; }
        } => {}
    }

    println!("✅ Rust collector shutdown complete");
    Ok(())
}
